package com.shujia.flink.window

import com.shujia.flink.window.Demo5TikTok.Event
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.ProcessingTimeSessionWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector


object Demo5TikTok {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val dataDS: DataStream[String] = env.socketTextStream("master", 8888)

    /**
      *
      * 1、将数据转换成样例类
      *
      */

    val eventDS: DataStream[Event] = dataDS.map(line => {
      val split: Array[String] = line.split(",")
      Event(split(0), split(1), split(2), split(3), split(4), split(5))
    })


    /**
      * 1、统计用户的行为
      * 将一个用户连续时间内的数据放在一起进行计算，铜用的类型偏好
      * 如果1分钟没有再刷新的视频为连续断开
      *
      */

    //安装用户表分组

    val keyByDS: KeyedStream[Event, String] = eventDS.keyBy(_.userId)

    //使用会话窗口
    val windowDS: WindowedStream[Event, String, TimeWindow] = keyByDS
      .window(ProcessingTimeSessionWindows.withGap(Time.seconds(5)))


    val userTypeScoreDS: DataStream[List[(String, String, Int)]] = windowDS.process(new TyprProcessWindowFunction)

    userTypeScoreDS.print()

    env.execute()


  }

  case class Event(userId: String, itemId: String, ifLike: String, ifEvaluate: String, ifRepetition: String, videoType: String)

}

class TyprProcessWindowFunction extends  ProcessWindowFunction[Event, List[(String, String, Int)], String, TimeWindow]{
  override def process(key: String,
                       context: Context,
                       events: Iterable[Event],
                       out: Collector[List[(String, String, Int)]]): Unit = {

    /**
      * 偏好打分规则
      * 点赞+1
      * 评价+1
      * 重复播放+1
      *
      */

    //计算每一条数据的得分
    val userEventScore: List[((String, String), Int)] = events
      .toList
      .map(event => {

        var score = 0

        if ("1".equals(event.ifLike)) {
          score += 1
        }

        if ("1".equals(event.ifEvaluate)) {
          score += 1
        }

        if ("1".equals(event.ifRepetition)) {
          score += 1
        }

        ((event.userId, event.videoType), score)

      })

    //对数据进行汇总
    val groupByMap: Map[(String, String), List[((String, String), Int)]] = userEventScore.groupBy(_._1)

    val userSumScore: List[(String, String, Int)] = groupByMap.toList.map {
      case (key: (String, String), scores: List[((String, String), Int)]) =>
        //计算每一个用户对每一个类型的总分
        val sumScore: Int = scores.map(_._2).sum
        (key._1, key._2, sumScore)
    }


    //将数据发送到下游
    out.collect(userSumScore)
  }
}
